package com.cet.reactor;

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class SchedulerDemo {

    public static void main(String[] args) throws IOException {

        // 流的发布默认使用当前线程
        Flux.range(1, 10)
                .log()
                .map(i -> i * 2)
                .log()
                .subscribe();

        System.out.println("-==================-");


        Flux.range(1, 10)
                .publishOn(Schedulers.immediate())
                .log()
                .subscribe();

        System.out.println("--------------------");

        Flux.range(1, 10)
                .publishOn(Schedulers.single())
                .log()
                .subscribe();

        System.out.println("-------------------------");

        Flux.range(1, 10)
//                .delayElements(Duration.ofSeconds(1))
                        .publishOn(Schedulers.newSingle("new"))
                                .log()
                                        .subscribe();

        System.out.println("----------------------");

        Flux.range(1, 10)
                        .publishOn(Schedulers.boundedElastic())
                                .log()
                                        .subscribe();

        System.out.println("----------------------");

        Flux.range(1, 10)
                        .publishOn(Schedulers.parallel())
                                .log()
                .map(i -> i * 2)
                .log()
                                        .subscribe();

        System.out.println("----------------------");

        Flux.range(1, 10)
                .log()
                .map(i -> i * 2)
                        .publishOn(Schedulers.fromExecutor(new ThreadPoolExecutor(8, 8, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>())))
                                .log()
                                        .subscribe();

        System.in.read();
    }
}
